package com.niodata.dp.plugin.yarn.log;

import java.net.InetAddress;
import java.util.Properties;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;


/**
 * Log collect v2.
 */
public class LogCollectV2 implements Runnable {

  private String logDir;
  private String appId;
  private String execId;
  private String taskName;
  private String jobName;
  private String hostName;
  private String hostIp;
  private String[] keys = new String[]{"logDir", "appId", "execId", "taskName", "jobName",
      "hostName",
      "hostIp"};

  private String[] values = null;

  private static final Integer DB_INDEX_YARN_LOG = 1;
  private static final String KEY_YARN_LOG_MAPPING = "YARN_LOG_INFO";

  private static String getEnvOrProperty(String key) {
    Properties sysProperties = System.getProperties();
    if (sysProperties.containsKey(key)) {
      return sysProperties.getProperty(key);
    } else if (System.getenv() != null && System.getenv().containsKey(key)) {
      return System.getenv(key);
    }
    return null;
  }

  /**
   * run.
   */
  @Override
  public void run() {
    logDir = getEnvOrProperty("spark.yarn.app.container.log.dir");
    String[] parts = logDir.split("/");
    appId = parts[parts.length - 2];
    String jobInfo = getEnvOrProperty("azkaban.link.job.url").split("\\?")[1];
    taskName = jobInfo.split("&")[0].split("=")[1];
    jobName = jobInfo.split("&")[2].split("=")[1];
    execId = getEnvOrProperty("azkaban.link.execution.url").split("=")[1];
    try {
      InetAddress addr = InetAddress.getLocalHost();
      hostIp = addr.getHostAddress();
      hostName = addr.getHostName();
    } catch (Exception e) {
      throw new RuntimeException(e.getMessage());
    }
    values = new String[]{logDir, appId, execId, taskName, jobName, hostName, hostIp};
    String logInfo = constructLogInfo();
    System.out.println(logInfo);

    String redisIp = getEnvOrProperty("redis.ip");
    String port = getEnvOrProperty("redis.port");
    String password = getEnvOrProperty("redis.password");
    if (redisIp != null && port != null) {
      int redisPort = Integer.parseInt(port);
      Jedis jedis = new Jedis(redisIp, redisPort);
      if (password != null) {
        jedis.auth(password);
      }
      try {
        jedis.select(DB_INDEX_YARN_LOG);
        jedis.lpush(KEY_YARN_LOG_MAPPING, logInfo);
        System.out.println("application info write to db success");
      } catch (Exception e) {
        e.printStackTrace();
      } finally {
        jedis.close();
      }
    } else {
      System.out.println("redis config not set in env");
    }
  }

  /**
   * constructLogInfo.
   */
  public String constructLogInfo() {
    try {
      StringBuffer sbu = new StringBuffer();
      for (int i = 0; i < keys.length; i++) {
        sbu.append(keys[i]).append("=").append(values[i]).append(";");
      }
      sbu.deleteCharAt(sbu.length() - 1);
      return sbu.toString();
    } catch (Exception e) {
      System.out.println(e.getMessage());
    }
    return null;
  }
}
